// SPDX-License-Identifier: MIT
// SPDX-FileCopyrightText: 2022 Ivan Baidakou

#include "rotor-light/supervisor.hpp"
#include "rotor-light/messages.hpp"
#include "rotor-light/planner.hpp"
#include "rotor-light/queue.hpp"

using namespace rotor_light;

SupervisorBase::SupervisorBase(char *backends_, size_t backends_count,
                               ActorBase **actors_,
                               details::ChildState *states_,
                               size_t actors_count_)
    : Parent(backends_, backends_count), actors{actors_}, states{states_},
      actors_count{actors_count_} {}

void SupervisorBase::start(bool poll_timer) {
  assert(state == State::off);
  initialize();
  send<message::ChangeState>(0, id, StateCommand::init);
  if (poll_timer) {
    assert(now && planner);
    send<message::RefreshTime>(0, id);
  }
}

void SupervisorBase::bind(Context &context, ActorId value) {
  bind(1 << value, this, context);
}

void SupervisorBase::initialize() {
  Parent::initialize();
  subscribe(&SupervisorBase::on_state_change_ack);
  subscribe(&SupervisorBase::on_refhesh_timer);
}

void SupervisorBase::on_state_change_ack(message::ChangeStateAck &message) {
  auto from = message.from;
  auto ack_state = message.state;
  auto ok = message.success;
  // self
  if (from == id) {
    if (ack_state == State::initialized) {
      if (ok) {
        start_actor(0);
      }
    }
    if (state == State::initializing && !ok) {
      send<message::ChangeState>(0, id, StateCommand::shutdown);
    }
    // children
  } else {
    for (size_t i = 1; i < actors_count; ++i) {
      auto child = actors[i];
      if (child->get_id() == from) {
        if (ack_state == State::initialized) {
          if (!ok) {
            on_child_init_failure(i);
          } else if (state == State::operational) {
            start_actor(i);
          }
        } else if (ack_state == State::off) {
          on_child_down(i);
        }

        if (state == State::initializing && ok) {
          if (check_child_state(State::initialized, true)) {
            Parent::advance_init();
          }
        }
      }
    }
  }
}

void SupervisorBase::on_refhesh_timer(message::RefreshTime &) {
  process_watchdog();
  send<message::RefreshTime>(0, id);
}

void SupervisorBase::on_child_init_failure(size_t actor_index) {
  auto child = actors[actor_index];
  states[actor_index - 1].down_state = State::initializing;
  send<message::ChangeState>(0, child->id, StateCommand::shutdown);
}

void SupervisorBase::on_child_down(size_t actor_index) {
  using Underlying = std::underlying_type_t<FailPolicy>;
  auto child = actors[actor_index];
  auto child_state = states[actor_index - 1];
  assert(child->get_state() == State::off);
  auto underlying_policy = static_cast<Underlying>(child->get_fail_policy());
  if (underlying_policy & static_cast<Underlying>(FailPolicy::restart)) {
    bool do_restart = (child_state.down_state == State::initializing) ||
                      (underlying_policy ==
                       static_cast<Underlying>(FailPolicy::force_restart));
    if (do_restart) {
      child_state.down_state = State::off;
      init_child(actor_index);
    }
  } else if (underlying_policy &
             static_cast<Underlying>(FailPolicy::escalate)) {
    bool do_stop = (child_state.down_state == State::initializing) ||
                   (underlying_policy ==
                    static_cast<Underlying>(FailPolicy::force_escalate));

    if (do_stop) {
      child_state.down_state = State::off;
      if (state == State::initializing) {
        send<message::ChangeStateAck>(0, supervisor->id, id, State::initialized,
                                      false);
      } else {
        stop();
      }
    }
  }

  if (state == State::shutting_down) {
    if (check_child_state(State::off, true)) {
      Parent::advance_stop();
    }
  }
}

bool SupervisorBase::check_child_state(State subject, bool skip_self) {
  size_t from = skip_self ? 1 : 0;
  size_t left = skip_self ? actors_count - 1 : actors_count;
  for (size_t i = from; i < actors_count; ++i) {
    auto actor = actors[i];
    auto actor_state = actor->get_state();
    if (subject != State::off) {
      if (actor_state >= subject) {
        --left;
      } else if (state == State::off) {
        if (actor->get_fail_policy() == FailPolicy::ignore) {
          --left;
        }
      }
    } else {
      if (actor_state == State::off) {
        --left;
      }
    }
  }
  return left == 0;
}

uint8_t SupervisorBase::bind(ActorId initial_value, SupervisorBase *supervisor,
                             Context &context) {
  queue = context.queue;
  planner = context.planner;
  now = context.now;
  auto counter = ActorBase::bind(initial_value, supervisor, context);
  for (size_t i = 1; i < actors_count; ++i) {
    auto value = initial_value << counter;
    counter += actors[i]->bind(value, this, context);
  }
  for (uint8_t i = 1; i <= counter; ++i) {
    mask |= initial_value << i;
  }

  return counter;
}

void SupervisorBase::advance_init() {
  for (size_t i = 1; i < actors_count; ++i) {
    init_child(i);
  }
  if (check_child_state(State::initializing, false)) {
    Parent::advance_init();
  }
}

void SupervisorBase::advance_start() {
  Parent::advance_start();
  for (size_t i = 1; i < actors_count; ++i) {
    auto actor = actors[i];
    if (actor->state == State::initialized) {
      start_actor(i);
    }
  }
}

void SupervisorBase::advance_stop() {
  for (size_t i = 1; i < actors_count; ++i) {
    auto child = actors[i];
    if (child->get_state() != State::off) {
      auto child_id = child->get_id();
      send<message::ChangeState>(0, child_id, StateCommand::shutdown);
    }
  }
  if (check_child_state(State::off, true)) {
    Parent::advance_stop();
  }
}

void SupervisorBase::process() {
NEXT:
  auto message = queue->next();
  while (message) {
    auto &msg = *message;
    dispatch(msg, msg.get_type_id());
    message.reset();
    message = queue->next();
  }
  if (now && process_watchdog()) {
    goto NEXT;
  }
}

size_t SupervisorBase::process_watchdog() {
  size_t events{0};
  auto now = this->now();
  auto it = planner->consume(now);
  while (it) {
    it.call_next();
    ++events;
  }
  return events;
}

void SupervisorBase::dispatch(Message &message, MessageTypeId message_type_id) {
  if (id & message.to) {
    auto ptr = reinterpret_cast<Backend *>(backends_ptr);
    auto end = ptr + backend_idx + 1;
    for (; ptr != end; ++ptr) {
      auto handler = static_cast<HandlerBase *>(ptr);
      if (handler->message_id == message_type_id) {
        handler->call(message, *this);
      }
    }
  }
  for (size_t i = 1; i < actors_count; ++i) {
    auto actor = actors[i];
    if (actor->mask & message.to) {
      if (actor->mask != actor->id) {
        auto sup = static_cast<SupervisorBase *>(actor);
        sup->dispatch(message, message_type_id);
      } else {
        auto ptr = reinterpret_cast<Backend *>(actor->backends_ptr);
        auto end = ptr + actor->backend_idx + 1;
        for (; ptr != end; ++ptr) {
          auto handler = static_cast<HandlerBase *>(ptr);
          if (handler->message_id == message_type_id) {
            handler->call(message, *actor);
            break;
          }
        }
      }
    }
  }
}

void SupervisorBase::init_child(size_t actor_index) {
  auto child = actors[actor_index];
  auto child_id = child->get_id();
  child->initialize();
  send<message::ChangeState>(0, child_id, StateCommand::init);
}

void SupervisorBase::start_actor(size_t actor_index) {
  auto actor = actors[actor_index];
  auto actor_id = actor->get_id();
  send<message::ChangeState>(0, actor_id, StateCommand::start);
}
